package skewreduce.framework.logical;

import java.util.IdentityHashMap;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;

import skewreduce.framework.EaggDriver;
import skewreduce.framework.EaggPlan;
import skewreduce.framework.physical.PhysicalOp;
import skewreduce.lib.Cube2;
import skewreduce.lib.Partition;



public abstract class LogicalOp implements Iterable<PhysicalOp> {
    public enum Type {
        SAMPLE,
        OPTIMIZE,
        REOPTIMIZE,
        PARTITION,
        GENERATE,
        WORK,
        MERGE,
        FINALIZE
    }

    protected final EaggDriver          driver;
    protected final Type          type;
    protected final int           numOutputs;
    private   IdentityHashMap<PhysicalOp.ID,PhysicalOp> phyOps;

    public LogicalOp(EaggDriver driver,Type t,int out) {
        this.driver = driver;
        this.type = t;
        this.numOutputs = out;
        phyOps = new IdentityHashMap<PhysicalOp.ID,PhysicalOp>();
    }

    public Type getType() { return type; }

    /**
     * @return number of output streams per physical instance
     */
    public int getNumOutputStreams() { return numOutputs; }
    public String getAppName() { return driver.getAppName(); }
    public EaggDriver getDriver() { return driver; }
    public Configuration getConfiguration() {
        return driver.getConf();
    }

    public Iterator<PhysicalOp> iterator() {
        return phyOps.values().iterator();
    }
    
    public final PhysicalOp createPhysicalOp(EaggPlan plan,Partition context) {
    	PhysicalOp op = createInstance();
    	op.initialize(plan,this,context);
    	phyOps.put( op.getID(), op );
    	return op;
    }
    
    protected abstract PhysicalOp createInstance();
}
